/**
 * GenServNetManTest.java
 * ************************** 
 * @date Apr 11, 2011
 * @author Michele Tamburini
 * 
 * **************************
 */
package JUnitTests;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import gossipServices.aggregation.AggActiveThread;
import gossipServices.aggregation.AggPassiveThread;
import gossipServices.aggregation.SingleValueContainer;
import gossipServices.basic.nodeDescriptors.ConcreteNodeDescriptor;
import gossipServices.basic.nodeDescriptors.NodeDescriptor;
import gossipServices.basic.view.ConcretePartialView;
import gossipServices.gossipMessages.AggregationMessage;
import gossipServices.gossipMessages.GeneralGossipMessage;
import gossipServices.gossipMessages.PssMessage;
import gossipServices.network.GeneralServicesMessageDispatcher;
import gossipServices.network.GeneralServicesNetManager;
import gossipServices.pss.ActivePSSThread;
import gossipServices.pss.PassivePSSThread;

import java.util.ArrayDeque;
import java.util.Queue;
import java.util.Vector;

import messages.Message;
import networkRmi.NetworkManager;

import org.junit.Before;
import org.junit.Test;

/**
 * Description: 
 *
 */
public class GenServNetManTest {
	
	class MyServiceNetMan extends GeneralServicesMessageDispatcher {

		/**
		 * @param networkManager
		 */
		public MyServiceNetMan(NetworkManager networkManager) {
			super(networkManager, AggregationMessage.class, 
					AggPassiveThread.class, AggActiveThread.class);
			nonBlockingReceiveWaitTime = 50;
		}
		
		public Vector<Message> getActiveThreadMessages(){
			return activeReceivedMessages;
		}
		
		public Vector<Message> getPassiveThreadMessages(){
			return passiveReceivedMessages;
		}
		
		public Vector<NodeDescriptor> getWaitingIncoming(){
			return waitingIncoming;
		}
		
		public Queue<Message> getRawMessages(){
			return rawReceivedMessages;
		}
		
	}

	private FakeNetMan fakeNetMan = new FakeNetMan();
	private MyServiceNetMan netMan;
	
	public static final int MESSAGES_NUMBER = 15;
	
	NodeDescriptor node1, node2, node3, node4;
	Message msgArray[] = new Message[MESSAGES_NUMBER];
	ArrayDeque<Message> messages = new ArrayDeque<Message>();
	ArrayDeque<Message> oldMessages;
	Vector<NodeDescriptor> waiting;
	int countAggregationActiveMessages = 0; 
	int countAggregationPassiveMessages = 0;
	

	/**
	 * @throws java.lang.Exception
	 */
	@Before
	public void setUp() throws Exception {
		node1 = new ConcreteNodeDescriptor("node1");
		node2 = new ConcreteNodeDescriptor("node2");
		node3 = new ConcreteNodeDescriptor("node3");
		node4 = new ConcreteNodeDescriptor("node4");
		
		netMan = new MyServiceNetMan(fakeNetMan);
		
		msgArray[0] = new AggregationMessage(node2, node1, new SingleValueContainer(new Double(0)), 
				AggActiveThread.class, 0, 50);
		msgArray[1] = new AggregationMessage(node2, node1, new SingleValueContainer(new Double(25)), 
				AggActiveThread.class, 0, 49);
		msgArray[2] = new AggregationMessage(node3, node1, new SingleValueContainer(new Double(0)), 
				AggActiveThread.class, 0, 50);
		msgArray[3] = new AggregationMessage(node4, node1, new SingleValueContainer(new Double(0)), 
				AggActiveThread.class, 0, 50);
		msgArray[4] = new GeneralGossipMessage();
		msgArray[5] = new GeneralGossipMessage(node3, node1);
		msgArray[6] = new PssMessage(new ConcretePartialView(0), node2, node1, ActivePSSThread.class);
		msgArray[7] = new PssMessage(new ConcretePartialView(0),node4, node1, ActivePSSThread.class);
		msgArray[8] = new PssMessage(new ConcretePartialView(0), node2, node1, PassivePSSThread.class);
		msgArray[9] = new AggregationMessage(node4, node1, new SingleValueContainer(new Double(15)), 
				AggPassiveThread.class, 78, 31);
		msgArray[10] = new AggregationMessage(node4, node1, new SingleValueContainer(new Double(7.5)), 
				AggPassiveThread.class, 78, 32);
		msgArray[11] = new AggregationMessage(node3, node1, new SingleValueContainer(new Double(30.5)), 
				AggPassiveThread.class, 78, 25);
		msgArray[12] = new AggregationMessage(node3, node1, new SingleValueContainer(new Double(15)), 
				AggPassiveThread.class, 78, 31);
		msgArray[13] = new AggregationMessage(node4, node1, new SingleValueContainer(new Double(10)), 
				AggPassiveThread.class, 78, 33);
		msgArray[14] = new AggregationMessage(node3, node1, new SingleValueContainer(new Double(10)), 
				AggPassiveThread.class, 78, 10);
		
		for(int i=0; i<MESSAGES_NUMBER; i++)
			messages.add(msgArray[i]);
		
		
		oldMessages = (ArrayDeque<Message>) messages.clone();
		
		for(Message m : messages){
			GeneralGossipMessage msg = (GeneralGossipMessage) m;
			try{
				if( (msg).getGossipThread().equals(AggActiveThread.class) )
					countAggregationActiveMessages++;
			}catch(NullPointerException e){
				//it isn't a Message with gossipThread set
			}
		}
	}
	
	@Test
	public void outputBeforeTesting(){
		printOut("**********************");
		printOut("Output before testing:");
		printOut("messages:");
		for(Message m : messages)
			printOut(m);
		
		printOut("oldMessages:");
		for(Message m: oldMessages)
			printOut(m);
		printOut("**********************");
	}

	@Test
	public void firstTest(){
		Queue<Message> netManRawMessages = netMan.getRawMessages();
		
		printOut("**********************");
		
		countAggregationMessages();
		
		printOut("found: "+countAggregationActiveMessages+" Aggregation messages");
		
		assertTrue(netManRawMessages.size() == 0);
		
		printOut("messages:");
		for(Message m : messages)
			printOut(m);
		
		netMan.extractIncomingMessages(messages);
		
		printOut("left raw messages: ");
		for(Message m : messages)
			printOut(m);
				
		
		printOut("Vectors:");
		printOut("rawMessages actual size: "+netManRawMessages.size());
		printOut("activeMessages actual size: "+netMan.getActiveThreadMessages().size());
		printOut("passiveMessages actual size: "+netMan.getPassiveThreadMessages().size());
		
		/*
		 * All messages have been dispatched to the active or the
		 * passive incoming message queues and the raw queue
		 * is now empty
		 */
		assertTrue(netManRawMessages.size() == 0);
		
		/*
		 * All the Messages coming from an Active thread are forwarded
		 * to the Passive received Messages queue by default.
		 */
		assertEquals(countAggregationActiveMessages, netMan.getPassiveThreadMessages().size());
		
		assertEquals(countAggregationPassiveMessages, netMan.getActiveThreadMessages().size());
		
		assertEquals( (oldMessages.size() - (countAggregationActiveMessages +  countAggregationPassiveMessages)), 
				messages.size());
		printOut("**********************");
	}
	


	@Test
	public void testReceiveNodeDescriptor() {
		printOut("**********************");
		printOut("TESTING: receive(NodeDescriptor node)");
		Message m;
		
		try {
			countAggregationMessages();		
			netMan.extractIncomingMessages(messages);

			assertEquals(countAggregationActiveMessages, netMan.getPassiveThreadMessages().size());
			assertEquals(countAggregationPassiveMessages, netMan.getActiveThreadMessages().size());

			m = netMan.receive(node2);
			assertTrue(m == null);
			assertEquals(countAggregationActiveMessages,netMan.getPassiveThreadMessages().size());
			assertEquals(1, netMan.getWaitingIncoming().size());

			NodeDescriptor target = node4;
			printOut("Extracting messages from: "+target);
			String prec = "";
			for(int i=0; i<3; i++){
				if(i>0)
					prec = m.getID();
				try {
					m = netMan.receive(target);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				printOut(m);
				assertTrue(m.getSender().equals(target));
				assertTrue(m.getID().equals(prec) == false);
			}


			target = node2; 
			Message incomingMsg = new AggregationMessage(target, node1, new SingleValueContainer(new Double(999)), 
					AggPassiveThread.class, 101, 33);
			messages.add(incomingMsg);
			netMan.extractIncomingMessages(messages);

			m = netMan.receive(target);
			assertEquals(0, netMan.getWaitingIncoming().size());
			assertTrue(m != null);
			assertEquals(incomingMsg.getID() ,m.getID());
			printOut("arrival simulation from: "+target+" >> "+m);


			target = node3;
			printOut("Extracting from "+ target);
			for(int i=0; i<3; i++){
				if(i>0)
					prec = m.getID();
				m = netMan.receive(target);
				printOut(m);
				assertTrue(m.getSender().equals(target));
				assertTrue(m.getID().equals(prec) == false);
			}

			netMan.receive(node2);
			netMan.receive(node3);
			netMan.receive(node4);
			assertEquals(3, netMan.getWaitingIncoming().size());

			assertEquals(0, netMan.getActiveThreadMessages().size());
			printOut("**********************");
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
//		fail("Not yet implemented");
	}

	@Test
	public void testReceive() {
		printOut("**********************");
		printOut("TESTING RECEIVE()");
		printOut("Original Messages:");
		printMessages();
		netMan.extractIncomingMessages(messages);
		
		try {
			/*
			 * verify that all messages are returned in the 
			 * same declaration order: 
			 */
			for(int i=0; i<countAggregationActiveMessages; i++){
				Message m;
				m = netMan.receive();
				printOut(m);
				assertTrue(m.getID().equals(msgArray[i].getID()));
			}

			// verify that the passive messages queue is now empty
			assertTrue(netMan.getPassiveThreadMessages().size() == 0);
			printOut("**********************");
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
//		fail("Not yet implemented");
	}
	
	@Test
	public void testPruneWatingIncoming(){
		printOut("**********************");
		printOut("TESTING PRUNE_WAITING_INCOMING()");
		int threshold = GeneralServicesNetManager.getBOUND_WAITINGINCOMING_SIZE();
		int age = GeneralServicesNetManager.getOLD_AGE_THRESHOLD() + 1;
		int offset[] = {3, 7, 8, GeneralServicesNetManager.getBOUND_WAITINGINCOMING_SIZE()+1};
		NodeDescriptor node = new ConcreteNodeDescriptor("sweetNode", age);
		
		try {
			for(int j = 0; j<offset.length; j++ ){
				printOut("TEST n "+j+") with offset: "+offset[j]);
				netMan.clearReceivedMessages();
				for(int i=0; i<threshold + offset[j]; i++){
					printOut(i+" add");
					netMan.receive(node);
				}
				printOut("Is waitingIncoming queue pruned??? let's see: ");
				printOut(netMan.getWaitingIncoming());
				printOut("expected size: "+offset+" actualSize: "+netMan.getWaitingIncoming().size());
				if(offset[j] == GeneralServicesNetManager.getBOUND_WAITINGINCOMING_SIZE())
					assertEquals(0, netMan.getWaitingIncoming().size());
				else
					assertEquals(offset[j]-1, netMan.getWaitingIncoming().size());
			}
		
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		printOut("**********************");
	}

	

	@Test
	public void testScanForMessageChecked() {
		fail("Not yet implemented");
	}

	@Test
	public void testScanForMessage() {
		fail("Not yet implemented");
	}

	private void printOut(Object s){
		System.out.println(s);
	}
	
	private void printMessages() {
		for(Message m : messages)
			printOut(m);
	}
	
	/**
	 * 
	 */
	private void countAggregationMessages() {
		countAggregationActiveMessages=0;
		countAggregationPassiveMessages = 0;
		for(Message m : messages){
			GeneralGossipMessage msg = (GeneralGossipMessage) m;
			try{
				if( (msg).getGossipThread().equals(AggActiveThread.class) )
					countAggregationActiveMessages++;
				if(msg.getGossipThread().equals(AggPassiveThread.class))
					countAggregationPassiveMessages++;
			}catch(NullPointerException e){
				//it isn't a Message with gossipThread set
			}
		}
	}
}

